/*******************************************************************/
/* @ All Rights Reserved Copyright (C) 2014 broada,Co.,Ltd.        */
/*                                                                 */
/*  NAME      = frtask_do.c                                        */
/*  NOTE      = 任务处理                                           */
/*  DATE      = 2014/10/23                                         */
/*******************************************************************/

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>

#include "frcomm.h"
#include "frlog.h"
#include "frmem.h"
#include "frstr.h"
#include "frsutl.h"

#include "conntrack.h"
#include "module.h"
#include "frame.h"
#include "frame_params.h"
#include "frct_comm.h"
#include "frct_mng.h"
#include "frct_expect.h"
#include "frtask_mng.h"
#include "frmodule_mng.h"
#include "frproto_detect.h"
#include "frreport_mng.h"
#include "frstat_system.h"
#include "frstat_thread.h"
#include "frstat_server.h"

/* response format最大重复次数 */
#define APM_L7_RESPONSE_MAX_TIME    100

/*******************************************************************************
*  FUNC     :  关联链路的处理
*  ARGS     :  p_st_report_data     (Frreport_data *,数据保存到该结构体)
*           :  p_st_ct              (Conntrack *)
*           :  p_st_data_in         (Analyz_data_in *, 协议判断用数据)
*  RTN      :
*  NOTE     :
*******************************************************************************/
//FR_STATIC void __frtask_do_relative(Conntrack *p_st_ct
//                                  , Analyz_data_in *p_st_data_in)
//{
//    L7_analyzer *p_st_l7_analyzer=NULL;
//    L7_module *p_st_l7_module=NULL;
//    u_short us_proto =0, us_port_srv=0;
//    u_int ui_ip_cli=0, ui_ip_srv=0;
//    Frreport_data st_report_data;
//    int i_ret_fun=0;
//
//    if(p_st_ct==NULL || p_st_data_in==NULL)
//        return;
//
//    p_st_l7_analyzer = frproto_l7_get_analyzer(p_st_ct);
//    if(p_st_l7_analyzer)
//        p_st_l7_module = p_st_l7_analyzer->p_st_l7_module;
//    if( p_st_l7_module == NULL || p_st_l7_module->l7_redirct_info == NULL)
//        return;
//    i_ret_fun = p_st_l7_module->l7_redirct_info(p_st_ct, &us_proto, &ui_ip_cli, &ui_ip_srv, &us_port_srv);
//    if(i_ret_fun == FR_FAIL)
//        return;
//
//    //加入到关联连接里
//    FRPRINT_L4(us_proto, ui_ip_cli, 0, ui_ip_srv, us_port_srv);
//    //i_ret_fun = frct_expect_add(us_proto, ui_ip_cli, ui_ip_srv, us_port_srv , p_st_ct);
//    //frdbg("frct_expect_add ret=%d\n", i_ret_fun);
//    //frct_expect_output(NULL);
//
//    memset(&st_report_data, 0x00, sizeof(Frreport_data));
//    st_report_data.em_data_type = FRREPORT_TYPE_CONN_EXPECT;
//    i_ret_fun = frreport_mng_add_conn_expect(&st_report_data, us_proto, ui_ip_cli, ui_ip_srv, us_port_srv
//                                           , p_st_ct->dst_ip, p_st_ct->dst_port, p_st_data_in->st_timestamp);
//    if(i_ret_fun == FR_FAIL)
//        return;
//
//    frreport_mng_send_report_data(&st_report_data);
//    frreport_free_report_data(&st_report_data);
//
//    return;
//}

/*******************************************************************************
*  FUNC     :  释放Analyz_data_in数据
*  ARGS     :  p_st_data_in       (Analyz_data_in *)
*  RTN      :
*  NOTE     :
*******************************************************************************/
FR_STATIC void __franalyz_data_free(Analyze_data_in *p_st_data_in)
{
    if(p_st_data_in == NULL)
        return;
    if(p_st_data_in->p_payload)
    {
        frfree(p_st_data_in->p_payload);
        p_st_data_in->p_payload = NULL;
    }

    return;
}

/*******************************************************************************
*  FUNC     :  第1层以后的协议判断与协议数据分析
*  ARGS     :  p_st_ct       (Conntrack *)
*           :  p_st_data_in  (Analyz_data_in *, 协议判断用数据)
*           :  p_st_data_out (Analyz_data_out *, 解析后出力数据)
*  RTN      :  l7_analyze的返回值
*  NOTE     :
*******************************************************************************/
FR_STATIC int __frtask_analyzer_next(Conntrack *p_st_ct
                                   , Analyze_data_in *p_st_data_in
                                   , Frhash_header  *p_st_hash_report
                                   , Frlist_header  *p_st_list_output)
{
    int i_ret_fun=0, i_ret_analyze=0;
    L7_analyzer *p_st_l7_analyzer=NULL;

    if(p_st_ct == NULL || p_st_data_in == NULL
        || p_st_hash_report==NULL || p_st_list_output == NULL )
        return APM_L7_ANALYZE_FAIL;

    //协议判断
    if(p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_IN_PROGRESS)
    {
        pthread_spin_lock(&p_st_ct->lock);
        //预先给协议分析生成一个节点
        if(frlist_node_idx(&p_st_ct->lst_hdr_l7proto, p_st_ct->i_active_l7proto_idx) == NULL )
            frct_l7proto_add_empty(p_st_ct);

        i_ret_fun = frproto_l7_detect_next(p_st_ct, p_st_data_in);
        if(i_ret_fun != FR_SUCC)
        {
            p_st_ct->i_flag_l7proto_status = APM_L7PROTO_STATUS_DONE;
            pthread_spin_unlock(&p_st_ct->lock);
            return APM_L7_ANALYZE_STOP;
        }
        pthread_spin_unlock(&p_st_ct->lock);
    }

    //协议分析
    i_ret_analyze = frproto_l7_analyze(p_st_ct, p_st_data_in, p_st_hash_report, p_st_list_output);
    //frdbg("data len=%d\tfrproto_l7_analyze ret=0x%04x\n", p_st_data_in->ui_payload_len, i_ret_analyze);

    if(i_ret_analyze == APM_L7_ANALYZE_TODO)
        return i_ret_analyze;

    //增加统计信息
    p_st_l7_analyzer = frproto_l7_get_analyzer(p_st_ct);
    if(p_st_l7_analyzer && p_st_l7_analyzer->p_st_l7_module)
    {
        //增加login logout的统计信息
        if(APM_L7_ANALYZE_IS_LOGIN(i_ret_analyze))
            frstat_system_module_session_add(p_st_l7_analyzer->p_st_l7_module->i_l7_proto, 1, 0);
        else if(APM_L7_ANALYZE_IS_LOGOUT(i_ret_analyze))
            frstat_system_module_session_add(p_st_l7_analyzer->p_st_l7_module->i_l7_proto, 0, 1);

        //增加服务节点的统计信息
        if(APM_L7_ANALYZE_IS_REQ(i_ret_analyze))
            frstat_server_add(p_st_ct->dst_ip, p_st_ct->dst_port, p_st_l7_analyzer->p_st_l7_module->i_l7_proto, 0, 1, 0);
        else if(APM_L7_ANALYZE_IS_RESP(i_ret_analyze))
            frstat_server_add(p_st_ct->dst_ip, p_st_ct->dst_port, p_st_l7_analyzer->p_st_l7_module->i_l7_proto, 0, 0, 1);
    }

    //判断程序是否结束
    if(i_ret_analyze == APM_L7_ANALYZE_FAIL || frlist_size(p_st_list_output) ==0 )
    {
        if(p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_IN_PROGRESS)
            p_st_ct->i_flag_l7proto_status = APM_L7PROTO_STATUS_DONE;
    }

    //增加线程统计信息
    if( p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_DONE
        && frct_l7proto_is_last(p_st_ct) == FR_SUCC )
    {
        if(i_ret_analyze == APM_L7_ANALYZE_FAIL)
            frstat_thread_add(pthread_self(), 0, 1, 0, 0);
        else if(APM_L7_ANALYZE_IS_REQ(i_ret_analyze) )
            frstat_thread_add(pthread_self(), 0, 0, 1, 0);
        else if(APM_L7_ANALYZE_IS_RESP(i_ret_analyze) )
            frstat_thread_add(pthread_self(), 0, 0, 0, 1);
    }

    return i_ret_analyze;
}

/*******************************************************************************
*  FUNC     :  Analyz_data_out节点内存释放
*  ARGS     :  p_st_node  (Frlist_node *)
*  RTN      :
*  NOTE     :
*******************************************************************************/
FR_STATIC void __list_output_node_free(Frlist_node *p_st_node)
{
    Analyze_data_out *p_st_cap_data=NULL;
    if(p_st_node==NULL || p_st_node->data == NULL )
        return;

    p_st_cap_data =  (Analyze_data_out*)FRLIST_GET_NODE_DATA(p_st_node);
    if(p_st_cap_data->p_payload)
    {
        frfree(p_st_cap_data->p_payload);
        p_st_cap_data->p_payload = NULL;
    }

    if(p_st_cap_data->p_st_request_fields)
    {
        frhashmap_varfield_free2(p_st_cap_data->p_st_request_fields);
        p_st_cap_data->p_st_request_fields = NULL;
    }

    if(p_st_cap_data->p_st_respones_fields)
    {
        frhashmap_varfield_free2(p_st_cap_data->p_st_respones_fields);
        p_st_cap_data->p_st_respones_fields = NULL;
    }

    return;
}

/*******************************************************************************
*  FUNC     :  Frreport_data节点内存释放
*  ARGS     :  p_st_node  (Frlist_node *)
*  RTN      :
*  NOTE     :
*******************************************************************************/
FR_STATIC void __list_report_node_free(Frlist_node *p_st_node)
{
    Frreport_data *p_st_report_data=NULL;
    if(p_st_node==NULL || p_st_node->data == NULL )
        return;

    p_st_report_data =  (Frreport_data*)FRLIST_GET_NODE_DATA(p_st_node);
    if(p_st_report_data)
    {
        frreport_free_report_data(p_st_report_data);
        frfree(p_st_report_data);
    }

    return;
}

/*******************************************************************************
*  FUNC     :  协议判断与协议数据分析
*  ARGS     :  p_st_ct (Conntrack *)
*           :  p_st_data_in  (Analyz_data_in *, 协议判断用数据)
*  RTN      :  
*  NOTE     :  内部使用p_st_ct->lock锁
*******************************************************************************/
FR_STATIC void __frtask_analyzer_first(Conntrack *p_st_ct
                                     , Analyze_data_in *p_st_data_in)
{
    int i_ret_fun=0, i_ret_analyze=0;
    int i_loop=0;
    Frhash_header st_hash_report;       /* 保存Frreport_data */
    Frlist_header *p_list_report=NULL;
    Frlist_header st_list_output_first; /* 保存Analyz_data_out */
    Frlist_header st_list_output_second;/* 保存Analyz_data_out */
    Frlist_node   *p_st_node_output_first=NULL;
    Frlist_node   *p_st_node_output_second=NULL;
    Frlist_node   *p_st_node_report=NULL;
    Frreport_data *p_st_report_data=NULL;
    Analyze_data_in *p_st_cap_data=NULL;
    L7_analyzer *p_st_l7_analyzer=NULL;

    char arr_buff_temp[128];
    
    if(p_st_ct == NULL || p_st_data_in == NULL)
        return ;
    if(frhash2_create(&st_hash_report, 2) != FR_SUCC)
    {
        FRLOG_ERROR_DETAIL("frhash2_create err");
        return;
    }
    frlist_header_init(&st_list_output_first);
    frlist_header_init(&st_list_output_second);

    pthread_spin_lock(&p_st_ct->lock);
    //第1层协议判断
    p_st_ct->i_active_l7proto_idx = 0;
    //有协议强制指定或用户指定的协议的更新,故已经识别过的协议再判断
    i_ret_fun = frproto_l7_detect_first(p_st_ct, p_st_data_in);
    //frwarn("frproto_l7_is_first = %d\n", i_ret_fun);
    pthread_spin_unlock(&p_st_ct->lock);
    if(i_ret_fun == FR_FAIL)
        goto END;
    if(p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_UNKNOWN
        || p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_FAIL )
        p_st_ct->i_flag_l7proto_status = APM_L7PROTO_STATUS_IN_PROGRESS;

    //协议分析
    i_ret_analyze = frproto_l7_analyze(p_st_ct, p_st_data_in, &st_hash_report, &st_list_output_first);
    //frdbg("data len=%d\tfrproto_l7_analyze ret=%d\n", p_st_data_in->ui_payload_len, i_ret_analyze);
    if(i_ret_analyze == APM_L7_ANALYZE_TODO)
        goto END;

    //增加统计信息
    p_st_l7_analyzer = frproto_l7_get_analyzer(p_st_ct);
    if(p_st_l7_analyzer && p_st_l7_analyzer->p_st_l7_module)
    {
        //增加login logout的统计信息
        if(APM_L7_ANALYZE_IS_LOGIN(i_ret_analyze))
            frstat_system_module_session_add(p_st_l7_analyzer->p_st_l7_module->i_l7_proto, 1, 0);
        else if(APM_L7_ANALYZE_IS_LOGOUT(i_ret_analyze))
            frstat_system_module_session_add(p_st_l7_analyzer->p_st_l7_module->i_l7_proto, 0, 1);

        //增加服务节点的统计信息
        if(APM_L7_ANALYZE_IS_REQ(i_ret_analyze))
            frstat_server_add(p_st_ct->dst_ip, p_st_ct->dst_port, p_st_l7_analyzer->p_st_l7_module->i_l7_proto, 0, 1, 0);
        else if(APM_L7_ANALYZE_IS_RESP(i_ret_analyze))
            frstat_server_add(p_st_ct->dst_ip, p_st_ct->dst_port, p_st_l7_analyzer->p_st_l7_module->i_l7_proto, 0, 0, 1);
    }

    //判断程序是否结束
    if(i_ret_analyze == APM_L7_ANALYZE_FAIL || frlist_size(&st_list_output_first) ==0 )
    {
        if(p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_IN_PROGRESS)
            p_st_ct->i_flag_l7proto_status = APM_L7PROTO_STATUS_DONE;
    }

    //增加线程统计信息
    if( p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_DONE
        && frct_l7proto_is_last(p_st_ct) == FR_SUCC)
    {
        if(i_ret_analyze == APM_L7_ANALYZE_FAIL)
            frstat_thread_add(pthread_self(), 0, 1, 0, 0);
        else if(APM_L7_ANALYZE_IS_REQ(i_ret_analyze) )
            frstat_thread_add(pthread_self(), 0, 0, 1, 0);
        else if(APM_L7_ANALYZE_IS_RESP(i_ret_analyze) )
            frstat_thread_add(pthread_self(), 0, 0, 0, 1);
    }

    //(协议解析失败)/(协议解析成功，但只有1个分析协议)，程序结束
    if(i_ret_analyze == APM_L7_ANALYZE_FAIL
       || (p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_DONE && frlist_size(&p_st_ct->lst_hdr_l7proto) == 1) )
        goto REPORT;

    /* 返回协议重新判断
     * 当前处理终止,上报已经解析的结果,释放当前协议开始的后续协议
     */
    if(i_ret_analyze == APM_L7_ANALYZE_REDETECT)
    {
        pthread_spin_lock(&p_st_ct->lock);
        frct_l7proto_remove(p_st_ct, 0);
        pthread_spin_unlock(&p_st_ct->lock);
        frct_l7proto_change_status(p_st_ct, APM_MODULE_STATUS_UNKNOWN, 0, APM_L7PROTO_STATUS_UNKNOWN);
        p_st_ct->i_flag_queue_redo =FR_YES;
        goto REPORT;
    }

    //如果有多个输出结果,则取第一个输出结果，进行协议的再次判断
    p_st_node_output_first = FRLIST_FIRST_NODE(&st_list_output_first);
    while(p_st_node_output_first)
    {
        //st_list_output保存的是Analyz_data_out,但与Analyz_data_in一致
        p_st_cap_data =  (Analyze_data_in*)FRLIST_GET_NODE_DATA(p_st_node_output_first);
        p_st_ct->i_active_l7proto_idx = 1;

        while(p_st_cap_data && p_st_ct->i_active_l7proto_idx < st_apm_params.ui_l7_deep_max )
        {
            //释放st_list_output_second
            frlist_remove_all(&st_list_output_second, __list_output_node_free);
            frlist_header_init(&st_list_output_second);

            i_ret_analyze = __frtask_analyzer_next(p_st_ct, p_st_cap_data, &st_hash_report, &st_list_output_second);
            __franalyz_data_free(p_st_cap_data);
            if(i_ret_analyze == APM_L7_ANALYZE_STOP)
                break;

            /* 返回协议重新判断
             * 当前处理终止,上报已经解析的结果,释放当前协议开始的后续协议
             */
            if(i_ret_analyze == APM_L7_ANALYZE_REDETECT)
            {
                pthread_spin_lock(&p_st_ct->lock);
                frct_l7proto_remove(p_st_ct, p_st_ct->i_active_l7proto_idx);
                pthread_spin_unlock(&p_st_ct->lock);
                frct_l7proto_change_status(p_st_ct, APM_MODULE_STATUS_UNKNOWN, 0, APM_L7PROTO_STATUS_IN_PROGRESS);
                break;
            }

            //取1个output数据节点
            p_st_node_output_second = FRLIST_FIRST_NODE(&st_list_output_second);
            //st_list_output保存的是Analyz_data_out,但与Analyz_data_in一致
            p_st_cap_data =  (Analyze_data_in*)FRLIST_GET_NODE_DATA(p_st_node_output_second);
            if(p_st_cap_data == NULL)
                break;

            //如果有多个出力数据，警告日志输出
            if(frlist_size(&st_list_output_second) > 1)
            {
                arr_buff_temp[0]='\0';
                frprint2_l4(arr_buff_temp, sizeof(arr_buff_temp)
                          , p_st_ct->ip_proto, p_st_cap_data->src_ip, p_st_cap_data->src_port
                          , p_st_cap_data->dst_ip, p_st_cap_data->dst_port);
                FRLOG_WARN_DETAIL("active_proto_idx=%d output_data>1 (count=%d);ct info[%s]\n", p_st_ct->i_active_l7proto_idx, frlist_size(&st_list_output_second), arr_buff_temp);
            }

            //设定为下一个协议节点
            p_st_ct->i_active_l7proto_idx += 1;
        }
        frlist_remove_all(&st_list_output_second, __list_output_node_free);

        // 如果超过深度,警告日志输出
        if(p_st_ct->i_active_l7proto_idx >= st_apm_params.ui_l7_deep_max)
        {
            if(p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_IN_PROGRESS)
                p_st_ct->i_flag_l7proto_status = APM_L7PROTO_STATUS_DONE;
            arr_buff_temp[0]='\0';
            frprint2_l4(arr_buff_temp, sizeof(arr_buff_temp)
                      , p_st_ct->ip_proto, p_st_cap_data->src_ip, p_st_cap_data->src_port
                      , p_st_cap_data->dst_ip, p_st_cap_data->dst_port);
            FRLOG_WARN_DETAIL("over max deep(%d); ct info[%s]", st_apm_params.ui_l7_deep_max, arr_buff_temp);
        }

        p_st_node_output_first = FRLIST_NEXT_NODE(p_st_node_output_first);
    }

REPORT:
    //处理上报数据
    FRHASH_LOOP(i_loop, &st_hash_report, p_list_report)
    {
        if(p_list_report == NULL)
            continue;
        p_st_node_report = FRLIST_FIRST_NODE(p_list_report);
        while(p_st_node_report)
        {
            p_st_report_data =  (Frreport_data*)FRLIST_GET_NODE_DATA(p_st_node_report);
            if(p_st_report_data)
                frreport_mng_send_report_data(p_st_report_data);
            p_st_node_report = FRLIST_NEXT_NODE(p_st_node_report);
        }
    }

END:
    frhash2_free(&st_hash_report, __list_report_node_free);
    frlist_remove_all(&st_list_output_first, __list_output_node_free);

    return ;
}

/*******************************************************************************
*  FUNC     :  Analyz_data_in处理完后确认是否需要保留
*  ARGS     :  p_st_data_in       (Analyz_data_in *)
*  RTN      :
*  NOTE     :
*******************************************************************************/
FR_STATIC void __frcap_queue_redo(Conntrack *p_st_ct, Analyze_data_in **pp_st_cap_data)
{
    Frlist_node *p_st_node_cap=NULL;
    Analyze_data_in *p_st_cap_data;

    if(p_st_ct == NULL
       || pp_st_cap_data == NULL || *pp_st_cap_data == NULL
       || p_st_ct->i_flag_queue_redo == FR_NO)
        return;

    p_st_cap_data = *pp_st_cap_data;
    //保存数据包到queue_hdr
    p_st_ct->i_flag_queue_redo = FR_NO;
    p_st_node_cap = frlist_node_create(p_st_cap_data);
    if(p_st_node_cap)
    {
        frlist_add_header(&p_st_ct->lst_cap_queue, p_st_node_cap);
    }
    else
    {
        __franalyz_data_free(p_st_cap_data);
        frfree(p_st_cap_data);
        FRLOG_ERROR_DETAIL("frlist_node_create err");
    }
    *pp_st_cap_data = NULL;
    return;
}

/*******************************************************************************
*  FUNC     :  Analyz_data_in处理完后确认是否需要保留
*  ARGS     :  p_st_data_in       (Analyz_data_in *)
*  RTN      :
*  NOTE     :
*******************************************************************************/
FR_STATIC void __frcap_queue_peek(Conntrack *p_st_ct, Analyze_data_in **pp_st_cap_data)
{
    Frlist_node *p_st_node_cap=NULL;
    Analyze_data_in *p_st_cap_data;

    if(p_st_ct == NULL
       || pp_st_cap_data == NULL || *pp_st_cap_data == NULL
       || p_st_ct->i_flag_cap_peek == FR_NO)
        return;

    p_st_cap_data = *pp_st_cap_data;

    //确认数据包是否需要保存
    if(p_st_ct->i_flag_cap_peek == FR_YES)
    {
        if(p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_IN_PROGRESS
            || p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_DONE)
            p_st_ct->i_flag_cap_peek = FR_NO;

        if(p_st_ct->ui_done_data_cnt > st_apm_params.ui_l7_detect_max_times )
            p_st_ct->i_flag_cap_peek = FR_NO;
    }

    if(p_st_ct->i_flag_cap_peek == FR_NO)
        return ;

    //保存数据包到queue_peek
    p_st_node_cap = frlist_node_create(p_st_cap_data);
    if(p_st_node_cap)
    {
        frlist_push(&p_st_ct->lst_cap_peek, p_st_node_cap);
    }
    else
    {
        __franalyz_data_free(p_st_cap_data);
        frfree(p_st_cap_data);
        FRLOG_ERROR_DETAIL("frlist_node_create err");
    }
    *pp_st_cap_data = NULL;
    return ;

}

/*******************************************************************************
*  FUNC     :  Analyz_data_in处理完后确认是否需要保留
*  ARGS     :  p_st_data_in       (Analyz_data_in *)
*  RTN      :
*  NOTE     :
*******************************************************************************/
FR_STATIC void __frcap_peek2queue(Conntrack *p_st_ct)
{
    Frlist_node *p_st_node_cap=NULL;

    if(p_st_ct == NULL
       || p_st_ct->i_flag_peek2queue == FR_NO
       || frlist_size(&p_st_ct->lst_cap_peek) <= 0 )
        return ;

    p_st_node_cap = FRLIST_FIRST_NODE(&p_st_ct->lst_cap_peek);
    if(p_st_node_cap)
    {
        frlist_merge_header(&p_st_ct->lst_cap_queue, &p_st_ct->lst_cap_peek);
        p_st_ct->i_flag_peek2queue = FR_NO;
    }

    return ;
}

/*******************************************************************************
*  FUNC     :  执行工作任务
*  ARGS     :  
*  RTN      :  
*  NOTE     :  
*******************************************************************************/
void *frtask_do(void *arg)
{
    Conntrack *p_st_ct = NULL;
    Frlist_node *p_st_node_cap=NULL;
    Analyze_data_in *p_st_cap_data=NULL;
    
    //FRLOG_DEBUG("frtask_do(%l) run...", pthread_self());
    FRLOG_DEBUG("frtask_do run...");
    frdbg("frtask_do(%lu) run...\n", pthread_self());
    while(g_i_flag_run)
    {
        //frdbg("frtask_get start\n");
        p_st_ct = frtask_get();
        //frdbg("frtask_get end(%p)\n", p_st_ct);
        if(p_st_ct == NULL)
        {
            usleep(10000);
            continue;
        }
        
        //获取解析数据并处理
        do
        {
            pthread_spin_lock(&p_st_ct->lock);
            p_st_node_cap = frlist_pop(&p_st_ct->lst_cap_queue);
            if(p_st_node_cap == NULL)
                p_st_cap_data = NULL;
            else
                p_st_cap_data = (Analyze_data_in *)FRLIST_GET_NODE_DATA(p_st_node_cap);
            pthread_spin_unlock(&p_st_ct->lock);
            if(p_st_node_cap)
                frfree(p_st_node_cap);
            if(p_st_cap_data == NULL)
                break;
            
            //FRPRINT_L4(p_st_ct->ip_proto, p_st_cap_data->src_ip, p_st_cap_data->src_port
            //             , p_st_cap_data->dst_ip, p_st_cap_data->dst_port);
            //FRPRINT_BIN("frtask_do", p_st_cap_data->p_data, p_st_cap_data->ui_len);

            //数据解析
            p_st_ct->i_flag_queue_redo =FR_NO;
            p_st_ct->ui_done_data_cnt +=1;
            __frtask_analyzer_first(p_st_ct, p_st_cap_data);
            p_st_ct->i_active_l7proto_idx= 0;

            //增加线程统计信息
            frstat_thread_add(pthread_self(), 1, 0, 0, 0);

            //确认数据是否需要重做
            if(p_st_ct->i_flag_queue_redo == FR_YES)
            {
                pthread_spin_lock(&p_st_ct->lock);
                __frcap_queue_redo(p_st_ct, &p_st_cap_data);
                //确认数据是否重新加入到解析队列
                __frcap_peek2queue(p_st_ct);
                pthread_spin_unlock(&p_st_ct->lock);
                //释放数据包
                if(p_st_cap_data)
                {
                    __franalyz_data_free(p_st_cap_data);
                    frfree(p_st_cap_data);
                    __sync_fetch_and_sub(&g_ul_pcap_queue_used, 1);
                }
                //释放保留数据
                if(frlist_size(&p_st_ct->lst_cap_peek) > 0
                   && (p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_IN_PROGRESS
                       || p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_FAIL
                       || p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_DONE)
                   )
                {
                    int i_cap_cnts= frlist_size(&p_st_ct->lst_cap_peek);
                    __sync_fetch_and_sub(&g_ul_pcap_queue_used, i_cap_cnts);
                    frlist_remove_all(&p_st_ct->lst_cap_peek, frcap_node_free);
                }
                continue;
            }

            //确认数据是否需要保留
            __frcap_queue_peek(p_st_ct, &p_st_cap_data);
            //释放保留数据
            if(frlist_size(&p_st_ct->lst_cap_peek) > 0
               && (p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_IN_PROGRESS
                       || p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_FAIL
                       || p_st_ct->i_flag_l7proto_status == APM_L7PROTO_STATUS_DONE)
               )
            {
                int i_cap_cnts= frlist_size(&p_st_ct->lst_cap_peek);
                __sync_fetch_and_sub(&g_ul_pcap_queue_used, i_cap_cnts);
                frlist_remove_all(&p_st_ct->lst_cap_peek, frcap_node_free);
            }

            //确认数据是否重新加入到解析队列
            if(p_st_ct->i_flag_peek2queue == FR_YES
               && frlist_size(&p_st_ct->lst_cap_peek) > 0 )
            {
                pthread_spin_lock(&p_st_ct->lock);
                __frcap_peek2queue(p_st_ct);
                pthread_spin_unlock(&p_st_ct->lock);
            }
            //释放数据包
            if(p_st_cap_data)
            {
                __franalyz_data_free(p_st_cap_data);
                frfree(p_st_cap_data);
                __sync_fetch_and_sub(&g_ul_pcap_queue_used, 1);
            }

        } while(g_i_flag_run);
        frct_put(p_st_ct);
        p_st_ct->flg_inTask = FR_NO;
    }
    
    pthread_exit(0);
    return (void*)NULL;
}

